package httpsrv

import (
	"bytes"
	"encoding/json"
	"net/http"
	"sync"
	"sync/atomic"
	"time"

	"gitee.com/JMArch/rpcmon-gateway/conf"
	"gitee.com/JMArch/rpcmon-gateway/log"
	"gitee.com/JMArch/rpcmon-gateway/route"
	"gitee.com/JMArch/rpcmon-gateway/storage"
	jwt "github.com/dgrijalva/jwt-go"
	"github.com/gorilla/websocket"
	"go.uber.org/zap"
)

var (
	secretKey = []byte("secret key")
)

type TokenResponse struct {
	Token string `json:"Token"`
}

type Connection struct {
	conn    *websocket.Conn
	wg      sync.WaitGroup
	done    int32
	exit    sync.Once
	errChan chan error
}

type HttpSrv struct {
	httpd      *http.Server
	log        *zap.Logger
	cert       string
	key        string
	websocket  *websocket.Upgrader
	cacheGroup *storage.CacheGroup
	quit       chan bool
	wg         sync.WaitGroup
	conns      sync.Map
	conf       *conf.GatewayConf
}

func New(cache *storage.CacheGroup, conf *conf.GatewayConf, logger *log.GatewayLog) *HttpSrv {

	srv := &HttpSrv{
		log:        logger.Named("httpsrv"),
		cert:       conf.Cert,
		key:        conf.Key,
		cacheGroup: cache,
		quit:       make(chan bool, 0),
		conf:       conf,
	}

	mux := route.New(srv)

	httpd := &http.Server{
		Addr:         conf.Addr,
		ReadTimeout:  2 * time.Second,
		WriteTimeout: 2 * time.Second,
		Handler:      mux,
	}
	srv.httpd = httpd

	srv.websocket = &websocket.Upgrader{
		ReadBufferSize:   1024,
		WriteBufferSize:  1024,
		HandshakeTimeout: 2 * time.Second,
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	}

	return srv
}

func (srv *HttpSrv) Serve() {

	srv.wg.Add(1)
	go func() {
		defer srv.wg.Done()

		var err error
		if srv.cert != "" && srv.key != "" {
			err = srv.httpd.ListenAndServeTLS(srv.cert, srv.key)
		} else {
			err = srv.httpd.ListenAndServe()
		}

		if err != nil {
			srv.log.Error("listen fail", zap.Error(err))
		}
	}()

	return
}

func (srv *HttpSrv) WebSocketHandler(w http.ResponseWriter, req *http.Request) {

	defer func() {
		if p := recover(); p != nil {
			srv.log.Error("websocket panic", zap.Any("error", p))
		}
	}()

	select {
	case <-srv.quit:
		w.WriteHeader(http.StatusServiceUnavailable)
		return
	default:
	}

	req.ParseForm()

	tokenStr := req.Form.Get("token")
	if tokenStr == "" {
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	token, err := jwt.ParseWithClaims(tokenStr, jwt.MapClaims{},
		func(token *jwt.Token) (interface{}, error) {
			return []byte(secretKey), nil
		})

	if err != nil || token.Valid == false {
		srv.log.Error("token invalid", zap.String("remote", req.RemoteAddr), zap.Error(err))
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	cacheType := req.Form.Get("type")
	if cacheType == "" {
		srv.log.Error("type is nil", zap.String("remote", req.RemoteAddr), zap.Error(err))
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	service := req.Form.Get("service")
	if service == "" {
		service = "_all_"
	}

	cache := srv.cacheGroup.Get(cacheType)
	if cache == nil {
		srv.log.Error("invalid cache type", zap.String("remote", req.RemoteAddr), zap.String("type", cacheType))
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	conn, err := srv.websocket.Upgrade(w, req, nil)
	if err != nil {
		srv.log.Error("upgrade websocket fail", zap.String("remote", req.RemoteAddr), zap.Error(err))
		w.WriteHeader(http.StatusServiceUnavailable)
		return
	}

	sConn := &Connection{
		conn:    conn,
		errChan: make(chan error, 0),
		done:    0,
	}

	srv.conns.Store(conn.RemoteAddr().String(), sConn)

	defer func() {
		srv.log.Debug("wait client websocket", zap.String("remote", req.RemoteAddr))
		select {
		case <-srv.quit:
		case <-sConn.errChan:
		}
		if atomic.CompareAndSwapInt32(&sConn.done, 0, 1) {
			sConn.conn.Close()
		}
		srv.conns.Delete(conn.RemoteAddr().String())
		sConn.wg.Wait()
		srv.log.Debug("close client websocket", zap.String("remote", req.RemoteAddr))
	}()

	srv.log.Debug("websocket connection", zap.String("remote", req.RemoteAddr), zap.Any("form", req.Form))
	srv.conns.Range(func(k, v interface{}) bool {
		srv.log.Debug("all conn list", zap.String("remote", k.(string)))
		return true
	})

	sConn.wg.Add(1)
	go func() {
		defer func() {
			if p := recover(); p != nil {
				srv.log.Error("websocket panic", zap.String("remote", conn.RemoteAddr().String()), zap.Any("error", p))
			}
			sConn.wg.Done()
		}()

		_, buf, err := conn.ReadMessage()
		if err != nil {
			srv.log.Error("websocket read fail", zap.String("remote", conn.RemoteAddr().String()), zap.Error(err))
			sConn.exit.Do(func() {
				close(sConn.errChan)
			})
			return
		}

		//srv.log.Debug("websocket read", zap.ByteString("message", buf), zap.String("remote", conn.RemoteAddr().String()), zap.Any("type", messageType))

		if bytes.Compare(buf, []byte("ping")) == 0 {
			err := conn.WriteMessage(websocket.TextMessage, []byte("pong"))
			if err != nil {
				srv.log.Error("websocket write fail", zap.String("remote", conn.RemoteAddr().String()), zap.Error(err))
				sConn.exit.Do(func() {
					close(sConn.errChan)
				})
				return
			}
		}
	}()

	sConn.wg.Add(1)
	go func() {
		defer func() {
			if p := recover(); p != nil {
				srv.log.Error("websocket panic", zap.String("remote", conn.RemoteAddr().String()), zap.Any("error", p))
			}
			sConn.wg.Done()
		}()

		resp, err := cache.Get(service, conn.RemoteAddr().String())
		if err != nil {
			srv.log.Error("cache get fail", zap.String("key", service), zap.Error(err))
			sConn.exit.Do(func() {
				close(sConn.errChan)
			})
			return
		}
		defer cache.Cancel(service, conn.RemoteAddr().String())

		for _, t := range resp.History {
			srv.log.Debug("send history to client", zap.Any("count", len(t.Data)), zap.String("remote", conn.RemoteAddr().String()))
			err := conn.WriteJSON(t.Data)
			if err != nil {
				srv.log.Error("send history fail", zap.String("key", service), zap.Error(err))
				sConn.exit.Do(func() {
					close(sConn.errChan)
				})
				return
			}
		}

		for {
			select {
			case <-srv.quit:
				srv.log.Error("send message fail, server closed", zap.String("remote", req.RemoteAddr))
				return
			case <-sConn.errChan:
				return
			case t := <-resp.Receive:
				if t == nil {
					srv.log.Error("send message fail, channel closed", zap.String("remote", req.RemoteAddr))
					return
				}
				srv.log.Debug("send message to client", zap.Any("count", len(t.Data)), zap.String("remote", conn.RemoteAddr().String()))
				err := conn.WriteJSON(t.Data)
				if err != nil {
					srv.log.Error("send message fail", zap.String("key", service), zap.Error(err))
					sConn.exit.Do(func() {
						close(sConn.errChan)
					})
					return
				}
			}
		}
	}()

	return
}

func (srv *HttpSrv) IndexHandler(w http.ResponseWriter, req *http.Request) {
	http.ServeFile(w, req, srv.conf.Index)
	return
}

func (srv *HttpSrv) GetTokenHandler(w http.ResponseWriter, req *http.Request) {

	defer func() {
		if p := recover(); p != nil {
			srv.log.Error("GetTokenHandler panic", zap.Any("error", p))
		}
	}()

	token := jwt.New(jwt.SigningMethodHS256)
	claims := make(jwt.MapClaims)
	claims["exp"] = time.Now().Add(srv.conf.TokenTTL).Unix()
	claims["iat"] = time.Now().Unix()
	token.Claims = claims

	tokenString, err := token.SignedString(secretKey)
	if err != nil {
		w.WriteHeader(http.StatusServiceUnavailable)
		srv.log.Error("make token string fail", zap.String("remote", req.RemoteAddr), zap.Error(err))
		return
	}

	response := &TokenResponse{
		Token: tokenString,
	}
	str, err := json.Marshal(response)
	if err != nil {
		w.WriteHeader(http.StatusServiceUnavailable)
		srv.log.Error("json marshal fail", zap.String("remote", req.RemoteAddr), zap.Error(err))
		return
	}

	w.Header().Set("Access-Control-Allow-Origin", "wx-mon.int.jumei.com")
	w.WriteHeader(http.StatusOK)
	w.Write(str)
	return
}

func (srv *HttpSrv) Close() {
	close(srv.quit)

	srv.httpd.Close()
	srv.conns.Range(func(k, v interface{}) bool {
		remote := k.(string)
		c := v.(*Connection)
		if atomic.CompareAndSwapInt32(&c.done, 0, 1) {
			c.conn.Close()
		}
		c.wg.Wait()
		c.exit.Do(func() {
			close(c.errChan)
		})
		srv.log.Debug("close client websocket", zap.String("remote", remote))
		return true
	})
	srv.wg.Wait()
}
